#include "asynch_return_queue.h"
#include "srv_asynch_man.h"
#include "Task_Base/CriticalSection.h"
#include "TcpPackBuf/TcpPackBuf.h"
#include <string.h>

enum ENUM_READ_OPERATION
{
    ENUM_READ_OPERATION_DEL_SOCKET,				//关闭套接字关联buf
    ENUM_READ_OPERATION_NORMAL,					//正常buf数据
};

typedef struct
{
    int                     code ;			//操作码
    int						socket;			//附加参数
}Read_Queue_Addition_Struct;

Asynch_Return_Queue::Asynch_Return_Queue(Srv_Asynch_Man *pAsynch_man):m_pAsynch_man(pAsynch_man)
{
    m_poCriSection_read_buf  = new CCriticalSection;
}

Asynch_Return_Queue::~Asynch_Return_Queue()
{
    close();
    delete m_poCriSection_read_buf;
}

void Asynch_Return_Queue::atq_add_service_handler(int client_socket)
{
    CCriticalAutoLock loAutoLock(*m_poCriSection_read_buf);

    map_socket_read_buf[client_socket] =new CTcpPackBuf ;
}

void Asynch_Return_Queue::atq_append_read_buf(int client_socket,const char *pBuf,int nlen)//当数据到来
{
    if( nlen == 0 )
        return ;

    Read_Queue_Addition_Struct read_queue_addition_struct ;
    read_queue_addition_struct.code = ENUM_READ_OPERATION_NORMAL;
    read_queue_addition_struct.socket = client_socket;

    std::string str_buf;
    str_buf.append((char*)&read_queue_addition_struct,sizeof(Read_Queue_Addition_Struct));  //加入buf前几个字节 附加结构
    str_buf.append(pBuf,nlen);				//数据接着加入

    putq(str_buf);		//入队
}

void Asynch_Return_Queue::atq_post_delete_socket(int client_socket)
{
    Read_Queue_Addition_Struct read_queue_addition_struct ;
    read_queue_addition_struct.code = ENUM_READ_OPERATION_DEL_SOCKET;
    read_queue_addition_struct.socket = client_socket;

    std::string str_buf;
    str_buf.append((char*)&read_queue_addition_struct,sizeof(Read_Queue_Addition_Struct));  //加入buf前几个字节 附加结构
    putq(str_buf);		//入队
}

int Asynch_Return_Queue::svc()
{
    char * szPackBuff = new char[4096];  //R_MAX_BUF ==4096

    std::string str ="";

    while(getq(str))	//出队
    {
        if (str=="")	//为空为退出标志
            return true;

        const char *pSocket		= str.data();
        int			buf_len		= str.size();

        Read_Queue_Addition_Struct *addition_struct = (Read_Queue_Addition_Struct *)pSocket ;

        switch(addition_struct->code)
        {

        case ENUM_READ_OPERATION_DEL_SOCKET:		//删除套接字关联buf
            {
                delete_service_handler( addition_struct->socket );
            }
            break;
        case ENUM_READ_OPERATION_NORMAL:			//正常处理数据
            {
                const char  *pBuf       = pSocket + sizeof(Read_Queue_Addition_Struct);//分离出数据
                int         nBufLen     = buf_len - sizeof(Read_Queue_Addition_Struct);//后续数据长度

                CTcpPackBuf * pRead_buf = find_socket_read_buf(addition_struct->socket);

                if(pRead_buf)
                {
                    pRead_buf->AppendData(pBuf,nBufLen);

                    memset(szPackBuff,0,4096);

                    int  lwAppDataLen = 0;

                    while(pRead_buf->UnPack(szPackBuff,4096,lwAppDataLen) == true)
                    {
                        if(m_pAsynch_man)
                            m_pAsynch_man->handle_read(addition_struct->socket,szPackBuff,lwAppDataLen);
                    }
                }
            }
            break;
        }
    }
    delete []szPackBuff;

    return 1;
}

void Asynch_Return_Queue::delete_service_handler( int client_socket )
{
    CCriticalAutoLock loAutoLock(*m_poCriSection_read_buf);

    std::map<int,CTcpPackBuf*>::iterator it_handle = map_socket_read_buf.find(client_socket);

    if (it_handle != map_socket_read_buf.end())
    {
        delete it_handle->second;

        map_socket_read_buf.erase(it_handle);
    }
}

CTcpPackBuf * Asynch_Return_Queue::find_socket_read_buf( int client_socket )
{
    CCriticalAutoLock loAutoLock(*m_poCriSection_read_buf);

    std::map<int,CTcpPackBuf*>::iterator it =map_socket_read_buf.find(client_socket);
    if (it != map_socket_read_buf.end())
    {
        return it->second;
    }
    return	NULL;
}

int Asynch_Return_Queue::close()
{
    std::string str("");
    int nthread  =thr_count();

    for (int i =0 ;i<nthread ;i++)
        putq(str);

    return 0;
}
